#1 获取输入参数个数，如果没有参数，直接退出
FILE_COUNT=$#
if((FILE_COUNT==0)); then
    echo "错误：未传递文件参数端口！"
    echo "用法：kafka.sh ip或host:2181（zookeeper连接地址，多个逗号隔开）"
    exit;
fi
# 开始下载
URL=https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.1/kafka_2.11-2.4.1.tgz
source ../linux/download_unzip.sh "kafka" $URL
# >>> source 调用本脚本时可用变量
# >>> $FILE_PATH 文件绝对路径：/opt/software/kafka_2.11-2.4.1.tgz
# >>> $FILE_EXTENSI 文件后缀：tgz
# >>> $File_HOME 软件HOME：/opt/module/kafka/kafka_2.11-2.4.1
# >>> $REPO_PATH 仓库目录：/opt/repository/kafka

# zookeeper连接地址
ZOOKEEPER_CONNECT=$1
# 创建日志目录
mkdir -p $File_HOME/logs
# 备份服务启动配置
if [ ! -f "$File_HOME/config/server.properties.bak" ]; then
    echo "第一次备份$File_HOME/config/server.properties"
    cp $File_HOME/config/server.properties $File_HOME/config/server.properties.bak
fi

# 配置kafka server 启动配置
sudo cat >$File_HOME/config/server.properties<<EOF
#broker的全局唯一编号，不能重复
broker.id=0
#删除topic功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的现成数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志存放的路径	
log.dirs=/opt/module/kafka/logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接Zookeeper集群地址
zookeeper.connect=$ZOOKEEPER_CONNECT
#单机版副本1
offsets.topic.replication.factor=1
EOF

# 配置KAFKA环境变量
if (( 0 == `cat /etc/profile | grep KAFKA_HOME | wc -l` )); then
sudo cat >>/etc/profile<<EOF

#===== 配置KAFKA_HOME环境变量 =====
export KAFKA_HOME=$File_HOME
export PATH=\$PATH:\$KAFKA_HOME/bin
EOF
fi


# 刷新环境变量
source /etc/profile
# 提示启动命令
echo "启动命令：$File_HOME/bin/kafka-server-start.sh -daemon $File_HOME/config/server.properties"
echo "停止命令：$File_HOME/bin/kafka-server-stop stop"

# 默认启动路径(后台启动，指定配置文件)
# 启动命令：/opt/module/kafka/kafka_2.11-2.4.1/bin/kafka-server-start.sh -daemon /opt/module/kafka/kafka_2.11-2.4.1/config/server.properties
# 停止命令：/opt/module/kafka/kafka_2.11-2.4.1/bin/kafka-server-stop.sh stop

# =========== Kafka命令行操作 ==============
# 1）查看当前服务器中的所有topic
# bin/kafka-topics.sh --zookeeper mini1:2181 --list

# 2）创建topic（选项说明：# --topic 定义topic名 --replication-factor  定义副本数 --partitions  定义分区数）
# bin/kafka-topics.sh --zookeeper mini1:2181 --create --replication-factor 1 --partitions 1 --topic first

# 3）删除topic（需要server.properties中设置delete.topic.enable=true否则只是标记删除。）
# bin/kafka-topics.sh --zookeeper mini1:2181 --delete --topic first

# 4）发送消息
# bin/kafka-console-producer.sh --broker-list mini1:9092 --topic first

# 5）消费消息（--from-beginning：会把主题中以往所有的数据都读取出来。）
# bin/kafka-console-consumer.sh --bootstrap-server mini1:9092 --from-beginning --topic first

# 6）查看某个Topic的详情
# bin/kafka-topics.sh --zookeeper mini1:2181 --describe --topic first
# 7）修改分区数
# bin/kafka-topics.sh --zookeeper mini1:2181 --alter --topic first --partitions 6

# =============== 单机无法消费 ================
# （1）在无意中查看了kafka的server.log日志，发现一直在报如下错误
# ERROR [KafkaApi-1001] Number of alive brokers '1' does not meet the required replication factor '3' for the offsets topic (configured via 'offsets.topic.replication.factor'). This error can be ignored if the cluster is starting up and not all brokers are up yet. (kafka.server.KafkaApis)
# （2）分析了下局面意思，大概就是broker的数量为2，但是server.properties配置的offsets.topic.replication.factor的值为3。看到这里我大概有两个解决方案。
# 一、添加broker的数目
# 二、修改server.properties中offsets.topic.replication.factor的值，改成2【Ambari中该值在kafka-broker中修改】

